﻿using System;
using System.Collections;
using System.Collections.Generic;
using System.Threading;
using ArgusLib;

namespace ArgusLib.Threading
{
	public static class Parallelization
	{
		const int SleepTime = 100;
		public delegate void ForEachBody<T>(T item, ThreadState threadState);
		public delegate void ForBody(int index, ThreadState threadState);
		public delegate IEnumerable<TaskBody> TaskBody(ThreadState threadState);

		public delegate void ProgressChangedHandler(double progress);
		public delegate void ProcessTasksProgressHandler(int tasksProcessed, int tasksToProcess);

		public static void ForEach<T>(IEnumerable collection, ForEachBody<T> body, ProgressChangedHandler progressChanged)
		{
			List<WaitHandle> waitHandles = new List<WaitHandle>();
			int count = 0;
			SynchronizedInt32 finished = new SynchronizedInt32(0);
			ThreadState threadState = new ThreadState();
			foreach (T item in collection)
			{
				count++;
				AutoResetEvent waitHandle = new AutoResetEvent(false);
				waitHandles.Add(waitHandle);
				ThreadPool.QueueUserWorkItem(new WaitCallback((arg)=>
					{
						if (threadState.CancelRequested == false)
						{
							body(item, threadState);
							if (progressChanged != null)
							{
								finished.Increment();
								progressChanged(finished.Object / (double)count);
							}
						}
						waitHandle.Set();
					}), null);
			}
			WaitHandle.WaitAll(waitHandles.ToArray());
		}

		public static void ForEach<T>(IEnumerable collection, ForEachBody<T> body)
		{
			Parallelization.ForEach<T>(collection, body, null);
		}

		public static void For(int count, ForBody body, ProgressChangedHandler progressChanged)
		{
			WaitHandle[] waitHandles = new WaitHandle[count];
			SynchronizedInt32 finished = new SynchronizedInt32(0);
			ThreadState threadState = new ThreadState();
			for (int i = 0; i < count; i++)
			{
				AutoResetEvent waitHandle = new AutoResetEvent(false);
				waitHandles[i] = waitHandle;
				int index = i;
				ThreadPool.QueueUserWorkItem(new WaitCallback((arg) =>
				{
					if (threadState.CancelRequested == false)
					{
						body(index, threadState);
						if (progressChanged != null)
						{
							finished.Increment();
							progressChanged(finished.Object / (double)count);
						}
					}
					waitHandle.Set();
				}), null);
			}
			WaitHandle.WaitAll(waitHandles);
		}

		public static void For(int count, ForBody body)
		{
			Parallelization.For(count, body, null);
		}

		public static void ProcessTasks(IEnumerable<TaskBody> tasks, int numberOfThreads, ProcessTasksProgressHandler progressChanged)
		{
			if (numberOfThreads < 1)
				throw new ArgumentOutOfRangeException("numberOfThreads");

			WaitHandle[] waitHandles = new WaitHandle[numberOfThreads];
			ThreadState threadState = new ThreadState();
			TaskCollection taskCollection = new TaskCollection(tasks);
			SynchronizedInt32 processed = new SynchronizedInt32(0);

			EventWaitHandle finished = new EventWaitHandle(false, EventResetMode.ManualReset);

			for (int i = 0; i < numberOfThreads; i++)
			{
				waitHandles[i] = new EventWaitHandle(false, EventResetMode.ManualReset);
			}

			for (int i = 0; i < numberOfThreads; i++)
			{
				int index = i;
				Thread thread = new Thread(new ThreadStart(() =>
				{
					EventWaitHandle waitHandle = (EventWaitHandle)waitHandles[index];
					while (true)
					{
						if (finished.WaitOne(0) == true)
							break;

						waitHandle.Reset();

						while (taskCollection.Count > 0)
						{
							if (threadState.CancelRequested == true)
								break;

							TaskBody task = taskCollection.TryPop();
							if (task == null)
								break;
							IEnumerable<TaskBody> newTasks = task(threadState);

							if (threadState.CancelRequested == true)
								return;

							if (newTasks != null)
								taskCollection.PushRange(newTasks);

							if (progressChanged != null)
							{
								processed.Increment();
								progressChanged(processed.Object, taskCollection.Count);
							}
						}
						waitHandle.Set();

						if (finished.WaitOne(0) == true)
							break;

						if (WaitHandle.WaitAll(waitHandles, 10) == true)
							break;
					}
					finished.Set();
				}));
				if (finished.WaitOne(0) == false)
					thread.Start();
			}
			finished.WaitOne();
		}

		private class TaskCollection
		{
			object _lock;
			Stack<TaskBody> tasks;

			public TaskCollection()
			{
				this._lock = new object();
				this.tasks = new Stack<TaskBody>();
			}

			public TaskCollection(int capacity)
			{
				this._lock = new object();
				this.tasks = new Stack<TaskBody>(capacity);
			}

			public TaskCollection(IEnumerable<TaskBody> tasks)
			{
				this._lock = new object();
				this.tasks = new Stack<TaskBody>(tasks);
			}

			public int Count { get { return this.tasks.Count; } }

			public void Push(TaskBody task)
			{
				lock (this._lock)
				{
					this.tasks.Push(task);
				}
			}

			public void PushRange(IEnumerable<TaskBody> tasks)
			{
				lock (this._lock)
				{
					foreach (TaskBody task in tasks)
					{
						this.tasks.Push(task);
					}
				}
			}

			public TaskBody Peek()
			{
				lock (this._lock)
				{
					return this.tasks.Peek();
				}
			}

			public TaskBody Pop()
			{
				lock (this._lock)
				{
					return this.tasks.Pop();
				}
			}

			public TaskBody TryPop()
			{
				lock (this._lock)
				{
					if (this.tasks.Count < 1)
						return null;
					else
						return this.tasks.Pop();
				}
			}
		}
	}
}
